package com.atguigu.gmall.realtime.common.base;

import com.atguigu.gmall.realtime.common.util.FlinkSourceUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;


/**
 * ClassName: BaseApp
 * Description:
 * 所有Flink应用的基类
 * 模板方法设计模式
 *      在父类中定义完成某一个功能的核算算法骨架（步骤），其中有些步骤在父类中没有办法实现，那么将其实现延迟到子类中
 *      好处：约定执行功能的模板
 *           在不改变核心骨架的前提下，每一个子类都可以有自己不同给的子类实现
 * @Create 2024/4/17 14:31
 */
public abstract class BaseApp {
    public void start(int port, int parallelism, String ckAndGroupId, String topic) {
        //TODO 1.基本环境准备
        //dim 10000 dwd 10010 dws 10020
        Configuration configuration = new Configuration();
        configuration.set(RestOptions.PORT,port);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(configuration);


        env.setParallelism(parallelism);


        //TODO 2.检查点相关设置
        CheckpointConfig checkpointConfig = env.getCheckpointConfig();
        //2.1 开启检查点
        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);

            //TODO 以下内容可不指定
        /*
        //2.2 设置检查点超时时间
        checkpointConfig.setCheckpointTimeout(60000L);
        //2.3 job取消后是否保留
        checkpointConfig.setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        //2.4 两个检查点间隔时间
        checkpointConfig.setMinPauseBetweenCheckpoints(2000L);
        //2.5 设置重启策略
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 3000L));
        //env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.days(30),Time.seconds(3)));
        //2.6 设置状态后端（指定存储路径）
        env.setStateBackend(new HashMapStateBackend());
        checkpointConfig.setCheckpointStorage("hdfs://hadoop102:8020/ck/" + ckAndGroupId);
        //2.7 设置操作Hadoop的用户
        System.setProperty("HADOOP_USER_NAME", "atguigu");
        */


        //TODO 3.从kafka主题读取数据
        String groupId = ckAndGroupId;
        //优化1
        KafkaSource<String> kafkaSource = FlinkSourceUtil.getKafkaSource(topic, groupId);
        //消费数据，封装为流
        DataStreamSource<String> kafkaSourceDS = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka_source");

        //TODO 4.处理逻辑
        handle(env,kafkaSourceDS);

        //TODO 5.提交作业
        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException();
        }
    }

    public abstract void handle(StreamExecutionEnvironment env, DataStreamSource<String> kafkaSourceDS);


}
